package com.isxcode.link.common.template;

import com.alibaba.fastjson.JSON;
import com.isxcode.link.common.constant.EventTypeConstants;
import com.isxcode.link.common.constant.KafkaConfigConstants;
import com.isxcode.link.common.pojo.LinkResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;

@Slf4j
public class LinkEventService {

    private final LinkEventHandler linkEventHandler;

    public LinkEventService(LinkEventHandler linkEventHandler) {

        this.linkEventHandler = linkEventHandler;
    }

    @KafkaListener(topics = KafkaConfigConstants.DEFAULT_TOPIC_NAME)
    public void listen(ConsumerRecord<String, String> record) {

        String executeId = record.key();
        LinkResponse linkResponse = JSON.parseObject(record.value(), LinkResponse.class);
        log.debug("监听到的返回数据 executeId：" + executeId + "starResponse:" + linkResponse.toString());

        switch (linkResponse.getLinkData().getEventType()) {

            case EventTypeConstants.QUERY_RESULT_EVENT:
                linkEventHandler.queryResultEvent(executeId, linkResponse);
                break;
            case EventTypeConstants.STATE_CHANGED_EVENT:
                linkEventHandler.stateChangedEvent(executeId, linkResponse);
                break;
            case EventTypeConstants.INFO_CHANGED_EVENT:
                linkEventHandler.infoChangedEvent(executeId, linkResponse);
                break;
            case EventTypeConstants.THREAD_ERROR_EVENT:
                linkEventHandler.threadErrorEvent(executeId, linkResponse);
                break;
        }
    }
}
